/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Collections;
import java.util.Comparator;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.TreeSet;

import cgl.narada.event.EventID;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.reliable.CompanionDbInfo;
import cgl.narada.service.reliable.events.RdCompanionEntityEvent;
import cgl.narada.service.reliable.events.RdNegotiationsServiceEvent;
import cgl.narada.service.reliable.events.RdRepublishedEntityEvent;
import cgl.narada.service.storage.StorageService;

/**
   This class is responsible to keep tracking of NBEvents and corresponding
   CompanionEvents

   @author Hasan Bulut
   $Date$
   $Revision$
 */
public class RdsEventPercolator {
    private String moduleName = "RdsEventPercolator2: ";
    private boolean debug = false;

    private Hashtable eventsReceived;
    private LinkedList companionInfos = new LinkedList();

    /** entityId <--> Hashtable*/
    private Hashtable entitiesAndCatenations;
    private EventProducer eventProducer;
    private ReliableDeliveryServiceImpl rdsService;
    private StorageService storageService;

    public RdsEventPercolator(ReliableDeliveryServiceImpl rdsService,
                              StorageService storageService) throws
            ServiceException {
        this.rdsService = rdsService;
        initializeProducer(rdsService);
        this.storageService = storageService;
        eventsReceived = new Hashtable();
        entitiesAndCatenations = new Hashtable();
        recoverFromFailure();
    }

    private void recoverFromFailure() throws ServiceException {
        int[] _clients = storageService.getListOfRegisteredEntities();

        if (_clients == null) {
            System.out.println(moduleName +
                               "No entities were previously registered"
                               + " ... no recovery needed");
            return;
        }

        TemplateInfo[] _templates = storageService.getListOfManagedTemplates();
        if (_templates == null) {
            System.out.println(moduleName +
                               "No templates were previously registered"
                               + "... no recovery needed");
            return;
        }

        if (_templates.length == 0) {
            System.out.println(moduleName +
                               "No templates were previously registered"
                               + "... no recovery needed ");
            return;
        }

        int _numOfTemplates = _templates.length;
        for (int i = 0; i < _numOfTemplates; i++) {
            int _templateId = _templates[i].getTemplateId();
            System.out.println(moduleName + "Recovering for TemplateID ["
                               + _templateId + "]");
            Object _templateKey = new Integer(_templateId);

            int[] _entities = storageService.getListOfRegisteredEntities(
                    _templateId);

            if (_entities == null) {
                continue;
            }

            if (_entities.length == 0) {
                continue;
            }

            for (int j = 0; j < _entities.length; j++) {
                int _entityId = _entities[j];
                Object _entityKey = new Integer(_entityId);
                if (debug) {
                    System.out.println("entityId: " + _entityId);
                }
                if (!entitiesAndCatenations.containsKey(_entityKey)) {
                    if (debug) {
                        System.out.println("New Table for entityId: " +
                                           _entityId);
                    }
                    entitiesAndCatenations.put(_entityKey, new Hashtable());
                }
                Hashtable templateCatenations = (Hashtable)
                                                entitiesAndCatenations
                                                .get(_entityKey);
                int _catenationNum = storageService.getCatenation(_templateId,
                        _entityId);
                EntityCompanionInfo entCmpInfo = new EntityCompanionInfo(
                        _entityId, _templateId);
                if (debug) {
                    System.out.println("Adding EntityCompanion for " +
                                       _templateId);
                }
                RdCompanionEntityEvent _pseudoEvent = new
                        RdCompanionEntityEvent(_entityId, _templateId,
                                               _catenationNum, 0, null);
                entCmpInfo.addToProcessedCatenations(_pseudoEvent);
                templateCatenations.put(_templateKey, entCmpInfo);
                System.out.println(moduleName +
                                   "Setting catenation number for "
                                   + "entity->" + _entityId +
                                   " for templateId->" + _templateId
                                   + " to [" + _catenationNum + "]");
                printEntityCatenations();
            }
            /* end for _entities */
        }
        /*end for _numOfTemplates */

        System.out.println(moduleName + "Recovery Complete ...");
    }

    private void initializeProducer(ReliableDeliveryServiceImpl rdsService) throws
            ServiceException {
        int rdEntityId = rdsService.getEntityId();
        ClientService clientService = SessionService.getClientService(
                rdEntityId);
        eventProducer = clientService.createEventProducer();
    }

    public void processCompanionEvent(RdCompanionEntityEvent companionEvent) {
        if (debug) {
            System.out.println(moduleName + "Received companion event \n"
                               + companionEvent);
        }
        EventID eventId = companionEvent.getEventId();
        EntityCompanionInfo entityCompanionInfo = getEntityCompanionInfo(
                companionEvent.getEntityId(), companionEvent.getTemplateId());
        if (entityCompanionInfo == null) {
            System.out.println("EntityCompanion is NULL");
            return;
        }

        /** Check to see if the companion should be discarded outright */
        if (entityCompanionInfo.shouldBeDiscarded(companionEvent)) {
            if (eventsReceived.containsKey(eventId)) {
                eventsReceived.remove(eventId);
            }
            if (debug) {
                System.out.println("CompanionEvent is discarded");
            }
            return;
        }
        if (eventsReceived.containsKey(eventId)) {
            entityCompanionInfo.processCompanionForExistingEvent(
                    companionEvent);
        } else {
            entityCompanionInfo.processCompanionForMissedEvent(
                    companionEvent);
        }
    }

    public boolean hasEventsAvailableForStorage() {
        if (eventsReceived.size() == 0) {
            System.out.println("Returning false eventsReceived.size() == 0");
            return false;
        }
        if (companionInfos.size() > 0) {
            int catNo = ((CompanionDbInfo) companionInfos.get(0)).
                        getCatenationNumber();
            if (catNo > 0) {
                return true;
            }
        }
        return false;
    }

    public CompanionDbInfo getReadyEventInfo() {
        if (debug) {
            System.out.println("companionInfos size: " + companionInfos.size());
        }
        for (int i = 0; i < companionInfos.size(); i++) {
            System.out.println(((CompanionDbInfo) companionInfos.get(i)).
                               toString());
        }
        return (CompanionDbInfo) companionInfos.removeFirst();
    }

    public NBEvent getReadyNBEvent(EventID eventId) {
        if (eventsReceived == null) {
            System.out.println("NULL eventsReceived");
        }
        if (eventId == null) {
            System.out.println("NULL eventId");
        }

        if (eventsReceived.containsKey(eventId)) {
            if (debug) {
                System.out.println("Returning event " + eventId);
            }
            return (NBEvent) eventsReceived.remove(eventId);
        } else {
            System.out.println("Returning NULL");
            return null;
        }
    }

    private EntityCompanionInfo getEntityCompanionInfo(int entityId,
            int templateId) {
        if ((entityId == 0) || (templateId == 0)) {
            System.out.println("(entityId == 0) || (templateId == 0)");
            return null;
        }
        Object entityKey = new Integer(entityId);
        Object templateKey = new Integer(templateId);
        if (debug) {
            System.out.println("EntityKey: " + entityKey + " templateKey: " +
                               templateKey);
        }
        if (!entitiesAndCatenations.containsKey(entityKey)) {
            if (debug) {
                System.out.println(
                        "Adding new Hashtable to entitiesAndCatenations");
            }
            entitiesAndCatenations.put(entityKey, new Hashtable());
        }
        Hashtable templateCatenations = (Hashtable) entitiesAndCatenations
                                        .get(entityKey);
        if (!templateCatenations.containsKey(templateKey)) {
            if (debug) {
                System.out.println("Adding new EntityCompanionInfo");
            }
            EntityCompanionInfo entCmpInfo = new EntityCompanionInfo(entityId,
                    templateId);
            templateCatenations.put(templateKey, entCmpInfo);
        }
        EntityCompanionInfo entCmpInfo = (EntityCompanionInfo)
                                         templateCatenations.get(templateKey);
        return entCmpInfo;
    }

    public void processPublishedEvent(NBEvent nbEvent) {
        EventID eventId = nbEvent.getEventHeaders().getEventId();
        if (eventId == null) {
            System.out.println(moduleName +
                               "Received event with NULL NBEvent id ");
            return;
        }

        if (eventsReceived.containsKey(eventId)) {
            System.out.println(moduleName + "Event with id = " + eventId
                               + "previously received");
            return;
        }

        eventsReceived.put(eventId, nbEvent);
        if (debug) {
            System.out.println(moduleName + "Received event with id \n" +
                               eventId);
        }

    }

    public void processRepublishedEvent(RdRepublishedEntityEvent
                                        republishedEvent) {
        if (debug) {
            System.out.println(moduleName + "REPUBLISHED EVENT RECEIVED WITH "
                               + "CATENATION NUMBER = "
                               +
                               republishedEvent.getCompanionEvent().
                               getCatenationNumber());
        }
        NBEvent nbEvent = republishedEvent.getPublishedEvent();
        RdCompanionEntityEvent companionEvent = republishedEvent
                                                .getCompanionEvent();

        processPublishedEvent(nbEvent);
        processCompanionEvent(companionEvent);
    }

    private void issueRepublishRequest(int entityId, int templateId,
                                       int catenation) {
        if (debug) {
            System.out.println(moduleName +
                               "Issuing republish request for catenation"
                               + " [" + catenation + "]");
        }
        boolean isAck = false;
        boolean isStandalone = false;
        issueNegotiationsEvent(entityId, templateId, catenation, isAck,
                               isStandalone);
    }

    private void issueNegotiationsEvent(int entityId, int templateId,
                                        int catenationNum, boolean isAck,
                                        boolean isStandalone) {

        RdNegotiationsServiceEvent negotiationEvent = new
                RdNegotiationsServiceEvent(entityId, templateId, isAck,
                                           isStandalone,
                                           catenationNum);
        try {
            NBEvent nbEvent = eventProducer.generateEvent(
                    TemplateProfileAndSynopsisTypes.STRING,
                    entityId + "/Negotiations",
                    negotiationEvent.getBytes());

            NBEventGenerator.setEventType(nbEvent,
                                          NBEventTypes.
                                          RELIABLE_DELIVERY_EXCHANGE);
            eventProducer.publishEvent(nbEvent);
        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        }
    }

    private void issueAcknowledgement(int entityId, int templateId,
                                      int catenation, boolean isStandalone) {
        try {
            storageService.advanceCatenation(catenation, templateId, entityId);
        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        }

        if (debug) {
            System.out.println(moduleName
                               + "Issuing acknowledgement for catenation [" +
                               catenation + "]");
        }

        boolean isAck = true;
        issueNegotiationsEvent(entityId, templateId, catenation, isAck,
                               isStandalone);
    }

    /** Removes an event from the release buffers after it has been
     stored to stable storage. We also proceed to issue acknowledgements
     regarding this storage */
    public void yankEventOffReleaseBuffers(CompanionDbInfo companionDbInfo) {
        int entityId = companionDbInfo.getEntityId();
        int templateId = companionDbInfo.getTemplateId();
        int catenation = companionDbInfo.getCatenationNumber();
        EntityCompanionInfo entityCompanionInfo = getEntityCompanionInfo(
                entityId, templateId);
        int result = entityCompanionInfo.getAckCatenation(catenation);
        boolean isStandalone = true;
        ;
        if (result != 0) {
            isStandalone = false;
            catenation = result;
        }
        issueAcknowledgement(entityId, templateId, catenation, isStandalone);
    }

    public void printEntityCatenations() {
        Enumeration enu = entitiesAndCatenations.keys();
        for (; enu.hasMoreElements(); ) {
            Integer ent = (Integer) enu.nextElement();
            System.out.println("EntityId: " + ent);
            Hashtable ht2 = (Hashtable) entitiesAndCatenations.get(ent);
            Enumeration enu2 = ht2.keys();
            for (; enu2.hasMoreElements(); ) {
                Integer temp = (Integer) enu2.nextElement();
                System.out.println("\tTemplateId: " + temp);
                EntityCompanionInfo entInfo = (EntityCompanionInfo) ht2.get(
                        temp);
                System.out.println("\t\t" + entInfo.toString());
            }
        }

    }

    class EntityCompanionInfo {
        TreeSet processedCompanions = new TreeSet(new CompanionComparator());
        // only catenation numbers
        List missedCatenations = Collections.synchronizedList(new LinkedList());
        /** catenationNumber <--> CompanionDbInfo */
        Hashtable companionDbInfos = new Hashtable();
        int entityId = 0;
        int templateId = 0;
        RepublishIssuer republishIssuer = new RepublishIssuer();
        int lastMissedCatenationReceived = 0;

        protected EntityCompanionInfo(int entityId, int templateId) {
            this.entityId = entityId;
            this.templateId = templateId;
            republishIssuer.start();
        }

        protected int getAckCatenation(int catenation) {
            int result = 0;
            if (missedCatenations.size() == 0) {
                result = catenation;
            } else {
                int firstMissedCatenation = Integer.parseInt(missedCatenations.
                        get(0).toString());
                if (catenation < firstMissedCatenation) {
                    result = firstMissedCatenation - 1;
                }
            }
            return result;
        }

        /**
         * If processedCompanions size is zero return false.
         * If processedCompanions contains it return true.
         * If the smallest catenation number of the processedCompanions is bigger
         * than the catenatin number of the comapnion event return true.
         * Else return false.
         * @param companionEvent RdCompanionEntityEvent
         * @return boolean
         */
        protected boolean shouldBeDiscarded(RdCompanionEntityEvent
                                            companionEvent) {
            if (processedCompanions.size() == 0) {
                return false;
            }
            if (processedCompanions.contains(companionEvent)) {
                return true;
            }
            RdCompanionEntityEvent cpEvent = (RdCompanionEntityEvent)
                                             processedCompanions.first();
            if (companionEvent.getCatenationNumber() <=
                cpEvent.getCatenationNumber()) {
                return true;
            }
            return false;
        }

        protected void addToProcessedCatenations(RdCompanionEntityEvent
                                                 companionEvent) {
            processedCompanions.add(companionEvent);
            if (debug) {
                System.out.println("Added to processedCompanions: " +
                                   companionEvent.getCatenationNumber() + " " +
                                   companionEvent.getTemplateId());
            }
        }

        protected void processCompanionForExistingEvent(RdCompanionEntityEvent
                companionEvent) {
            int companionCatNumber = companionEvent.getCatenationNumber();
            if (debug) {
                System.out.println("Processing for existing event " +
                                   companionEvent.getCatenationNumber());
            }
            if (companionEvent.getCatenationNumber() == 1) {
                CompanionDbInfo cmpInfo = null;
                if (companionDbInfos.containsKey("" + companionCatNumber)) {
                    cmpInfo = (CompanionDbInfo) companionDbInfos.get("" +
                            companionCatNumber);
                    cmpInfo.setUpdate(true);
                } else {
                    cmpInfo = rdsService.getCompanionDbInfo(companionEvent.
                            getTemplateId());
                    cmpInfo.setEntityId(companionEvent.getEntityId());
                    cmpInfo.setTemplateId(companionEvent.getTemplateId());
                    cmpInfo.setCatenationNumber(companionEvent.
                                                getCatenationNumber());
                    cmpInfo.setUpdate(false);
                }
                cmpInfo.setStatus(1);
                cmpInfo.setEventId(companionEvent.getEventId());
                companionInfos.add(cmpInfo);
                processedCompanions.add(companionEvent);
                rdsService.eventReadyForStorage();
                return;
            }
            int initCatNum = 0;
            if (debug) {
                System.out.println("processedCompanions size: " +
                                   processedCompanions.size());
            }
            if (processedCompanions.size() > 1) {
                initCatNum = ((RdCompanionEntityEvent) processedCompanions.
                              last()).getCatenationNumber();
            } else if (processedCompanions.size() == 1) {
                initCatNum = ((RdCompanionEntityEvent) processedCompanions.
                              first()).getCatenationNumber();
            }
            if (initCatNum > companionEvent.getCatenationNumber()) {
                // get info from companionDbInfos
                String catKey = "" + companionEvent.getCatenationNumber();
                if (!companionDbInfos.containsKey(catKey)) {
                    // should not come here
                    System.out.println("Error!! Companion:" + catKey);
                    return;
                }
                CompanionDbInfo cmpInfo = (CompanionDbInfo)
                                          companionDbInfos.get(catKey);
                // store event
                cmpInfo.setEventId(companionEvent.getEventId());
                cmpInfo.setEntityId(companionEvent.getEntityId());
                cmpInfo.setUpdate(true);
                cmpInfo.setStatus(1);
                if (debug) {
                    System.out.println("Setting CompanionDbInfo: " +
                                       cmpInfo.toString());
                }
                companionInfos.add(cmpInfo);
                // remove it from missed catNums
                companionDbInfos.remove(catKey);
                missedCatenations.remove(catKey);
                processedCompanions.add(companionEvent);
                trimProcessedCompanions();
                rdsService.eventReadyForStorage();
                return;
            }
            for (int i = initCatNum + 1; i < companionEvent.getCatenationNumber();
                         i++) {
                if (companionDbInfos.containsKey("" + i)) {
                    continue;
                }
                CompanionDbInfo cmpInfo = rdsService.getCompanionDbInfo(
                        companionEvent.getTemplateId());
                cmpInfo.setEntityId(companionEvent.getEntityId());
                cmpInfo.setTemplateId(companionEvent.getTemplateId());
                cmpInfo.setCatenationNumber(i);
                cmpInfo.setUpdate(false);
                cmpInfo.setStatus(0);
                companionInfos.add(cmpInfo);
                companionDbInfos.put("" + i, cmpInfo);
                missedCatenations.add("" + i);
                rdsService.eventReadyForStorage();
            }
            CompanionDbInfo cmpInfo = rdsService.getCompanionDbInfo(
                    companionEvent.getTemplateId());
            cmpInfo.setEntityId(companionEvent.getEntityId());
            cmpInfo.setTemplateId(companionEvent.getTemplateId());
            cmpInfo.setEventId(companionEvent.getEventId());
            cmpInfo.setCatenationNumber(companionEvent.getCatenationNumber());
            cmpInfo.setUpdate(false);
            cmpInfo.setStatus(1);
            companionInfos.add(cmpInfo);
            processedCompanions.add(companionEvent);
            trimProcessedCompanions();
            rdsService.eventReadyForStorage();
        }

        protected void trimProcessedCompanions() {
            boolean result = true;
            LinkedList resultSet = new LinkedList();
            Iterator it = processedCompanions.iterator();
            RdCompanionEntityEvent cpEvt1 = null;
            RdCompanionEntityEvent cpEvt2 = null;
            if (it.hasNext()) {
                cpEvt1 = (RdCompanionEntityEvent) it.next();
            } else {
                return;
            } while (result && it.hasNext()) {
                cpEvt2 = (RdCompanionEntityEvent) it.next();
                if (cpEvt1.getCatenationNumber() ==
                    (cpEvt2.getCatenationNumber() - 1)) {
                    RdCompanionEntityEvent cpEvt = cpEvt1;
                    resultSet.add(cpEvt);
                } else {
                    result = false;
                }
                cpEvt1 = cpEvt2;
            }
            for (int i = 0; i < resultSet.size(); i++) {
                processedCompanions.remove(resultSet.get(i));
            }
            resultSet.clear();
        }

        protected void processCompanionForMissedEvent(RdCompanionEntityEvent
                companionEvent) {
            if (debug) {
                System.out.println("Processing for missed event");
            }
            int companionCatNumber = companionEvent.getCatenationNumber();
            if (companionDbInfos.containsKey("" + companionCatNumber)) {
                if (debug) {
                    System.out.println("Missed event already in the list.");
                }
                return;
            }
            // issueRepublishRequest
            int entityId = companionEvent.getEntityId();
            int templateId = companionEvent.getTemplateId();
            int catenation = companionEvent.getCatenationNumber();

            issueRepublishRequest(entityId, templateId, catenation);
        }

        public String toString() {
            String str = "*****";
            if (processedCompanions.size() > 0) {
                str = processedCompanions.first().toString();
            }
            return str;
        }

        class RepublishIssuer extends Thread {
            protected boolean running = true;
            long sleepTime = 2000;
            public void run() {
                int catNum = 0;
                while (running) {
                    // send republish request for each missed event
                    for (int i = 0; i < missedCatenations.size(); i++) {
                        catNum = Integer.parseInt(missedCatenations.get(i).
                                                  toString());
                        if (debug) {
                            System.out.println("Issue Republish for entityId: " +
                                               entityId + " templateId: " +
                                               templateId + " catNum: " +
                                               catNum);
                        }
                        issueRepublishRequest(entityId, templateId, catNum);
                    }
                    // sleep for some time
                    try {
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException ex) {

                    }
                }
            }
        }

        class CompanionComparator implements Comparator {
            public int compare(Object one, Object two) {
                RdCompanionEntityEvent comOne = (RdCompanionEntityEvent) one;
                RdCompanionEntityEvent comTwo = (RdCompanionEntityEvent) two;
                return (comOne.getCatenationNumber() -
                        comTwo.getCatenationNumber());
            }

        }
    }

}
